package com.navinfo.platform.examples.parquet;

import com.navinfo.platform.qingqi.protocol.common.InterEventData;
import com.navinfo.platform.qingqi.protocol.java.EventDataPb;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.*;

import java.util.Arrays;
import java.util.List;

/**
 * PB转为普通BEAN模式
 */
public class MyParquetPBJavaSample {
    public static void main(String[] args) throws AnalysisException {
        // $example on:init_session$
        SparkSession spark = SparkSession
                .builder()
                .master("local[4]")
                .appName("MyParquetPBJavaSample")
                .getOrCreate();
        // $example off:init_session$

        runBasicDataFrameExample(spark);

        spark.stop();
    }

    private static void runBasicDataFrameExample(SparkSession spark) {
        // Create a JavaSparkContext using the SparkSession's SparkContext object
        JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());


        EventDataPb.EventData ed = EventDataPb.EventData.newBuilder().setTerminalId(10001L).setDuration(1000L).
                setEvent(EventDataPb.EventType.brake).setStartGpsTime(1L).setStopGpsTime(2L).build();

        InterEventData ied = new InterEventData();
        ied.setTerminalId(ed.getTerminalId());
        ied.setDuration(ed.getDuration());
        ied.setEvent(ed.getEvent().toString());
        ied.setStartGpsTime(ed.getStartGpsTime());
        ied.setStopGpsTime(ed.getStopGpsTime());

        List<InterEventData> data = Arrays.asList(ied);
        JavaRDD<InterEventData> distData = jsc.parallelize(data);

        // Apply a schema to an RDD of JavaBeans to get a DataFrame
        Dataset<Row> eventDF = spark.createDataFrame(distData, InterEventData.class);

        eventDF.show();

        eventDF.write().mode(SaveMode.Overwrite).partitionBy("terminalId").parquet("/user/root/event.parquet");

        Dataset<Row> eventReadDF = spark.read().parquet("/user/root/event.parquet");

        eventReadDF.show();

        eventReadDF.selectExpr("terminalId", "duration+1").show();
    }
}
